package com.cskaoyan.mall.search.mq;


import com.cskaoyan.mall.mq.constant.MqTopicConst;
import com.cskaoyan.mall.search.service.SearchService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.util.List;

@Component
public class ProductOnSaleConsumer {

    DefaultMQPushConsumer consumer;

    @Value("${rocketmq.namesrv.addr}")
    String nameserverAddress;

    @Value("${rocketmq.consumer.group}")
    String consumerGroup;

    @Autowired
    SearchService searchService;

    @PostConstruct
    public void init() {
        // 创建consumer对象
        consumer = new DefaultMQPushConsumer(consumerGroup + "_onsale");

        // 设置nameserver地址
        consumer.setNamesrvAddr(nameserverAddress);

        try {
            // 订阅主题
            consumer.subscribe(MqTopicConst.PRODUCT_ONSALE_TOPIC, "*");

            // 设置消息监听器
            consumer.setMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                    // 获取上架消息
                    MessageExt message = msgs.get(0);

                    // 获取消息体数据
                    byte[] body = message.getBody();
                    try {
                        String skuIdStr = new String(body, 0, body.length, "utf-8");
                        long skuId = Long.parseLong(skuIdStr);
                        // 调用service方法实现上架
                        searchService.upperGoods(skuId);
                    } catch (Exception e) {
                      e.printStackTrace();
                      return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            // 启动consumer
            consumer.start();
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
    }
}
